

namespace Stee.CAP8.MessageBus.Kafka.Configurations
{
    using System;
    using System.Collections.Generic;
    using Confluent.Kafka;
    using Stee.CAP8.MessageBus.Interface.Utility;
    using Stee.CAP8.MessageBus.Kafka.ResourceLib;

    /// <summary>
    /// Configuration for Kafka consumer
    /// See below reference for detailed documentation on indivdual configurations
    /// https://docs.confluent.io/current/installation/configuration/consumer-configs.html
    /// </summary>
    public class SubscriberConfiguration
    {
        #region Properties

        /// <summary>
        /// Gets the initial bootstrap servers' addresses as a CSV list of broker host or host:port.
        /// Mandatory configuration.
        /// </summary>
        public string BootstrapServers { get; set; } = string.Empty;

        /// <summary>
        /// Gets the client group id string.
        /// All clients sharing the same group.id belong to the same group.
        /// default: '' importance: high
        /// Mandatory configuration.
        /// </summary>
        public string GroupID { get; set; } = null;

        /// <summary>
        /// Gets or sets the client identifier
        /// </summary>
        public string ClientID { get; set; } = null;

        /// <summary>
        /// Gets or sets the cction to take when there is no initial offset in offset store or the desired
        /// offset is out of range: 'smallest','earliest' - automatically reset the offset
        /// to the smallest offset, 'largest','latest' - automatically reset the offset to
        /// the largest offset, 'error' - trigger an error which is retrieved by consuming
        /// messages and checking 'message->err'.
        /// default: largest importance: high
        /// </summary>
        public AutoOffsetReset AutoOffsetReset { get; set; } = AutoOffsetReset.Latest;

        /// <summary>
        /// Gets or sets the flag to automatically and periodically commit offsets in the background.
        /// Note: Setting this to false does not prevent the consumer from fetching previously committed
        /// start offsets. To circumvent this behaviour set specific start offsets per partition
        /// in the call to assign().
        /// default: true importance: high
        /// </summary>
        public bool EnableAutoCommit { get; set; } = true;

        /// <summary>
        /// Gets or sets the frequency in milliseconds that the consumer offsets are committed (written) to offset storage.
        /// Set as 0 to disable. This setting is used by the high-level consumer.
        /// default: 5000 importance: medium
        /// </summary>
        public int AutoCommitIntervalMs { get; set; } = 5000;

        /// <summary>
        /// Gets or sets the flag to automatically store offset of last message provided to application.
        /// The offset store is an in-memory store of the next offset to (auto-)commit for each partition.
        /// default: true importance: high
        /// </summary>
        public bool EnableAutoOffsetStore { get; set; } = true;

        /// <summary>
        /// Gets or sets the maximum allowed time between calls to consume messages (e.g., rd_kafka_consumer_poll())
        /// for high-level consumers. If this interval is exceeded the consumer is considered
        /// failed and the group will rebalance in order to reassign the partitions to another
        /// consumer group member. Warning: Offset commits may be not possible at this point.
        /// Note: It is recommended to set `enable.auto.offset.store=false` for long-time
        /// processing applications and then explicitly store offsets (using offsets_store())
        /// *after* message processing, to make sure offsets are not auto-committed prior
        /// to processing has finished. The interval is checked two times per second. See
        /// KIP-62 for more information.
        /// default: 300000 importance: high
        /// </summary>
        public int MaxPollIntervalMs { get; set; } = 300000;

        /// <summary>
        /// Gets or sets the client group session and failure detection timeout. The consumer sends periodic
        /// heartbeats (heartbeat.interval.ms) to indicate its liveness to the broker. If
        /// no hearts are received by the broker for a group member within the session timeout,
        /// the broker will remove the consumer from the group and trigger a rebalance. The
        /// allowed range is configured with the **broker** configuration properties `group.min.session.timeout.ms`
        /// and `group.max.session.timeout.ms`. Also see `max.poll.interval.ms`.
        /// default: 10000 importance: high
        /// </summary>
        public int SessionTimeoutMs { get; set; } = 10000;

        /// <summary>
        /// Gets or sets the flag to verify CRC32 of consumed messages, ensuring no on-the-wire or on-disk corruption to
        /// the messages occurred. This check comes at slightly increased CPU usage.
        /// default: false importance: medium
        /// </summary>
        public bool CheckCrcs { get; set; } = false;

        /// <summary>
        /// Gets or sets the name of partition assignment strategy to use when elected group leader assigns partitions to group members.
        /// default: range importance: medium
        /// </summary>
        public PartitionAssignmentStrategy PartitionAssignmentStrategy { get; set; } = PartitionAssignmentStrategy.Range;

        /// <summary>
        /// Gets the validity of subscriber configuration
        /// </summary>
        public bool IsValid
        {
            get { return this.Validate(); }
        }

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the SubscriberConfiguration class
        /// </summary>
        /// <param name="bootstrapServers">Kafka broker servers (bootstrap servers)</param>
        /// <param name="groupID">Consumer group ID</param>
        public SubscriberConfiguration(string bootstrapServers, string groupID)
        {
            bootstrapServers = bootstrapServers?.Trim();
            groupID = groupID?.Trim();

            if (string.IsNullOrEmpty(bootstrapServers))
            {
                throw new ArgumentNullException(nameof(bootstrapServers));
            }
            else if (string.IsNullOrEmpty(groupID))
            {
                throw new ArgumentNullException(nameof(groupID));
            }

            this.BootstrapServers = bootstrapServers;
            this.GroupID = groupID;
        }

        /// <summary>
        /// Initializes a new instance of the SubscriberConfiguration class.
        /// Make certain that all cases of dictionary keys matches that of the respective class properties.
        /// </summary>
        /// <param name="configurations">Configuration mappings to property name. Eg: "BootstrapServers" => "127.0.0.1:9092"</param>
        public SubscriberConfiguration(Dictionary<string, string> configurations)
        {
            if (configurations == null)
            {
                throw new ArgumentNullException(nameof(configurations));
            }
            else if (!configurations.ContainsKey(nameof(this.BootstrapServers)))
            {
                throw new ArgumentNullException($"{Kafka_General.KAFKA_NULLEMPTY_MANDATORY_CONFIGURATION}: {nameof(this.BootstrapServers)}");
            }
            else if (!configurations.ContainsKey(nameof(this.GroupID)))
            {
                throw new ArgumentNullException($"{Kafka_General.KAFKA_NULLEMPTY_MANDATORY_CONFIGURATION}: {nameof(this.GroupID)}");
            }

            //// Check mandatory configurations
            string bootstrapServers = configurations[nameof(this.BootstrapServers)];
            string groupID = configurations[nameof(this.GroupID)];

            if (string.IsNullOrEmpty(bootstrapServers))
            {
                throw new ArgumentNullException($"{Kafka_General.KAFKA_NULLEMPTY_MANDATORY_CONFIGURATION}: {nameof(this.BootstrapServers)}");
            }
            else if (string.IsNullOrEmpty(groupID))
            {
                throw new ArgumentNullException($"{Kafka_General.KAFKA_NULLEMPTY_MANDATORY_CONFIGURATION}: {nameof(this.GroupID)}");
            }

            this.InitializeProperties(configurations);
        }

        /// <summary>
        /// Initializes a new instance of the SubscriberConfiguration class
        /// </summary>
        /// <param name="currentConfig">Current subscriber configuration</param>
        private SubscriberConfiguration(SubscriberConfiguration currentConfig)
        {
            this.AutoCommitIntervalMs = currentConfig.AutoCommitIntervalMs;
            this.AutoOffsetReset = currentConfig.AutoOffsetReset;
            this.EnableAutoCommit = currentConfig.EnableAutoCommit;
            this.EnableAutoOffsetStore = currentConfig.EnableAutoOffsetStore;
            this.PartitionAssignmentStrategy = currentConfig.PartitionAssignmentStrategy;
            this.BootstrapServers = currentConfig.BootstrapServers;
            this.CheckCrcs = currentConfig.CheckCrcs;
            this.ClientID = currentConfig.ClientID;
            this.GroupID = currentConfig.GroupID;
            this.MaxPollIntervalMs = currentConfig.MaxPollIntervalMs;
            this.SessionTimeoutMs = currentConfig.SessionTimeoutMs;
        }

        /// <summary>
        /// Initializes the configuration mappings into properties
        /// </summary>
        /// <param name="configurations">Configuration mappings to property name. Eg: "BootstrapServers" => "127.0.0.1:9092"</param>
        private void InitializeProperties(Dictionary<string, string> configurations)
        {
            foreach (KeyValuePair<string, string> item in configurations)
            {
                switch (item.Key)
                {
                    case nameof(this.BootstrapServers):
                        this.BootstrapServers = item.Value?.Trim();
                        break;

                    case nameof(this.GroupID):
                        this.GroupID = item.Value?.Trim();
                        break;

                    case nameof(this.ClientID):
                        this.ClientID = item.Value?.Trim();
                        break;

                    case nameof(this.AutoOffsetReset):
                        this.AutoOffsetReset = ValueConverter.ConvertToEnum<AutoOffsetReset>(item.Value, this.AutoOffsetReset);
                        break;

                    case nameof(this.EnableAutoCommit):
                        this.EnableAutoCommit = ValueConverter.ConvertToBoolean(item.Value, this.EnableAutoCommit);
                        break;

                    case nameof(this.AutoCommitIntervalMs):
                        this.AutoCommitIntervalMs = ValueConverter.ConvertToInteger(item.Value, this.AutoCommitIntervalMs);
                        break;

                    case nameof(this.EnableAutoOffsetStore):
                        this.EnableAutoOffsetStore = ValueConverter.ConvertToBoolean(item.Value, this.EnableAutoOffsetStore);
                        break;

                    case nameof(this.MaxPollIntervalMs):
                        this.MaxPollIntervalMs = ValueConverter.ConvertToInteger(item.Value, this.MaxPollIntervalMs);
                        break;

                    case nameof(this.SessionTimeoutMs):
                        this.SessionTimeoutMs = ValueConverter.ConvertToInteger(item.Value, this.SessionTimeoutMs);
                        break;

                    case nameof(this.CheckCrcs):
                        this.CheckCrcs = ValueConverter.ConvertToBoolean(item.Value, this.CheckCrcs);
                        break;

                    case nameof(this.PartitionAssignmentStrategy):
                        this.PartitionAssignmentStrategy = ValueConverter.ConvertToEnum<PartitionAssignmentStrategy>(item.Value, this.PartitionAssignmentStrategy);
                        break;
                        
                    default:
                        //// TODO: log invalid property
                        break;
                }
            }
        }

        #endregion Constructors & Initializers


        #region Public Methods

        /// <summary>
        /// Clone a new instance of SubscriberConfiguration class
        /// </summary>
        /// <returns>New SubscriberConfiguration instance</returns>
        public SubscriberConfiguration Clone()
        {
            return new SubscriberConfiguration(this);
        }

        #endregion Public Methods


        #region Private Methods

        /// <summary>
        /// Validate subscriber configurations
        /// </summary>
        /// <returns>Return TRUE if configuration is valid else FALSE</returns>
        private bool Validate()
        {
            bool isValid = true;

            if (string.IsNullOrEmpty(this.BootstrapServers) ||
                string.IsNullOrEmpty(this.GroupID) ||
                this.MaxPollIntervalMs < 1000 ||
                this.SessionTimeoutMs < 1000)
            {
                isValid = false;
            }

            return isValid;
        }

        #endregion Private Methods
    }
}
